package com.javashitang.rabbitmq.chapter_7_publisherConfirm;

import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

/**
 * <p>
 *  注意这里的成功回调并不代表消费者已经消费，
 *  只是代表消息被投递到所有匹配的队列，
 *
 *  一旦消息被投递到所有匹配的队列之后，broker就会发送一个确认给生产者(包含消息的唯一ID)，
 *  这就使得生产者知道消息已经正确到达目的队列了，
 *  如果消息和队列是可持久化的，那么确认消息会在将消息写入磁盘之后发出，broker回传给生产者的确认消息中delivery-tag域包含了确认消息的序列号，
 *  此外broker也可以设置basic.ack的multiple域，表示到这个序列号之前的所有消息都已经得到了处理；
 * </p>
 *
 * @author gobigg 
 * @since 2020/1/17
 */
@Slf4j
public class ConfirmProducer {

    public static final String EXCHANGE_NAME = "confirm_exchange";

    public static void main(String[] args)
            throws IOException, TimeoutException, InterruptedException {
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("192.168.99.100");
        connectionFactory.setPort(5672);
        connectionFactory.setUsername("admin");
        connectionFactory.setPassword("admin");

        Connection connection = connectionFactory.newConnection();
        Channel channel = connection.createChannel();
        channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT);

        // 启用发布者确认模式
        channel.confirmSelect();

        String routingKey = "error";
        for (int i = 0; i < 1000000000; i++) {
            String message = "hello rabbitmq " + i;
            channel.basicPublish(EXCHANGE_NAME, routingKey, null, message.getBytes());
            // 一条一条确认，返回为true，则表示发送成功
            if (channel.waitForConfirms()) {
                log.info("send success, routingKey: {}, message: {}", routingKey ,message);
            } else {
                log.info("send fail, routingKey: {}, message: {}", routingKey ,message);
            }
        }
        channel.close();
        connection.close();
    }
}
